
-- source
CREATE TABLE hdp_lbg_supin_ods_zp_58_zzdetail_visitor (
    msg string
)
with(
'connector' = 'kafka',
'topic' = ' hdp_lbg_supin_ods_zp_58_zzdetail',
'properties.bootstrap.servers' = 'bh-kafka3-1.58dns.org:9092,bh-kafka3-2.58dns.org:9092,bh-kafka3-3.58dns.org:9092,bh-kafka3-4.58dns.org:9092,bh-kafka3-5.58dns.org:9092',
'properties.group.id' = 'hdp_lbg_supin_ods_zp_58_zzdetail_visitor-dev',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_supin_ods_zp_58_zzdetail-FRZfA',
'scan.startup.mode' = 'group-offsets',
'sink.parallelism' = '16',
'format' = 'raw'
);

set table.dynamic-table-options.enabled = true;

create function SplitFunction AS 'udtf.SplitFunction';

create function Split001Udf as 'com.bigdata.udf.Split001Udf';

create function TjfromFormatUDF as 'com.bigdata.udf.TjfromFormatUDF';


select
    arr[5] as pid,
    arr[6] as sid,
    arr[16] as imei,
    arr[19] as imei_md5,
    arr[35] as tjfrom,
    TjfromFormatUDF(arr[35]) as tjfrom2,
    arr[38] as info_id,
    arr[39] as info_buid,
    arr[40] as info_title,
    arr[46] as info_type,
    arr[47] as info_types,
    arr[54] as info_enterprise_name,
    arr[58] as zpsrc
from (
select Split001Udf(regex_str) as arr
from hdp_lbg_supin_ods_zp_58_zzdetail_visitor /*+ OPTIONS('scan.startup.mode'='latest-offset') */
,LATERAL TABLE(SplitFunction(msg,'\n'))
) t;



